package com.shujia.core

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.api.scala._

object Demo4EventTime {

  def main(args: Array[String]): Unit = {

    /**
      * 时间
      *
      * 处理时间
      * 事件时间
      * 接收时间
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //当所有线程中时间到达窗口的结束时间才开始计算
    env.setParallelism(1)
    //处理时间 TimeCharacteristic.ProcessingTime
    //TimeCharacteristic.EventTime   时间时间
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val ds = env.socketTextStream("node1", 8888)

    val eventDS = ds.map(line => {
      val split = line.split(",")

      Event(split(0), split(1).toLong)
    })

    //统计最近5秒每个车出现的次数

    eventDS
      .assignAscendingTimestamps(_.time) // 指定时间字段   毫秒级别
      .map(event => (event.id, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(5))
      .reduce((x, y) => (x._1, x._2 + y._2))
      .print()

    env.execute("Demo4EventTime")


  }

  case class Event(id: String, time: Long)

}
